package com.qfedu.edu.iotserver.service.impl;

import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.qfedu.edu.bo.QueueMessage;
import com.qfedu.edu.constance.RedisConstance;
import com.qfedu.edu.exception.handler.BusinessException;
import com.qfedu.edu.manager.RabbitMQManager;
import com.qfedu.edu.manager.RedisManager;
import com.qfedu.edu.mqtt.data.MessageHeader;
import com.qfedu.edu.pojo.ExceptionBusiness;
import com.qfedu.edu.iotserver.mapper.ExceptionBusinessMapper;
import com.qfedu.edu.iotserver.service.ExceptionBusinessService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.qfedu.edu.result.ResponseCode;
import com.qfedu.edu.utils.StringUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import sun.security.provider.certpath.BuildStep;

import java.io.UnsupportedEncodingException;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author xiaobobo
 * @since 2024-09-13
 */
@Service
public class ExceptionBusinessServiceImpl extends ServiceImpl<ExceptionBusinessMapper, ExceptionBusiness> implements ExceptionBusinessService {

    @Autowired
    private ExceptionBusinessMapper exceptionBusinessMapper;

    @Autowired
    private RabbitMQManager rabbitMQManager;

    @Autowired
    private RedisManager redisManager;

    @Override
    public void handlerExceptionBusinessInfo(String val) {
        //就是前面我们为什么要将发送的队列中的消息进行封装呢?  QueueMessage
        QueueMessage queueMessage = JSON.parseObject(val, QueueMessage.class);
        //封装插入到数据库的数据
        ExceptionBusiness exceptionBusiness = getExceptionBusiness(val, queueMessage);
        //在插入记录之前要解决幂等性问题
        Integer mqttId = exceptionBusiness.getMqttId();
        //通过mqttId查询数据是否已经存在
        //这里其实就是解决消息的幂等性问题 防止队列消息重复消费
        QueryWrapper<ExceptionBusiness> exceptionBusinessQueryWrapper = new QueryWrapper<>();
        exceptionBusinessQueryWrapper.eq("mqtt_id", mqttId);
        ExceptionBusiness exceptionBusiness1 = exceptionBusinessMapper.selectOne(exceptionBusinessQueryWrapper);
        if (null == exceptionBusiness1) {
            //说明这个记录在表中还不存在....
            //然后向数据库插入数据
            exceptionBusinessMapper.insert(exceptionBusiness);
        }
    }

    @Override
    public IPage<ExceptionBusiness> findAll(Integer currentPage, Integer pageSize) {
        if (null == currentPage || null == pageSize) {
            throw new BusinessException(ResponseCode.PARAMETER_EXCEPTION);
        }

        IPage<ExceptionBusiness> page = new Page<>();
        page.setCurrent(currentPage);
        page.setSize(pageSize);

        //执行到这里 说明参数无误
        return exceptionBusinessMapper.selectPage(page, null);
    }

    @Override
    public void retry(String id) throws UnsupportedEncodingException {
        if (StringUtils.isEmpty(id)) {
            throw new BusinessException(ResponseCode.PARAMETER_EXCEPTION);
        }
        //执行到这里 说明参数无误
        //重试?
        //查询数据库查询到消息
        ExceptionBusiness exceptionBusiness = exceptionBusinessMapper.selectById(id);
        //取出交换机名字
        String exchangeName = exceptionBusiness.getExchangeName();
        String routingKey;
        if(!StringUtils.isEmpty(exchangeName)){
            //说明有交换机的名字
            routingKey=exceptionBusiness.getRoutingKey();
            //有交换机的那么发送消息只能发送到交换机
            rabbitMQManager.sendToExchange(exchangeName,routingKey,exceptionBusiness.getMessage());
        }else{
            if(!StringUtils.isEmpty(exceptionBusiness.getQueueName())){
                //说明交换机没有名字  那么就发送消息到队列
                rabbitMQManager.sendToWork(exceptionBusiness.getQueueName(),exceptionBusiness.getMessage() );
            }else{
                throw new BusinessException(ResponseCode.QUEUE_NAME_EXCEPTION);
            }
        }
    }

    @Override
    public Integer findQueueMessageRetryStatus(String mqttId) {
        //这个消息只能去Redis中查询
        if(StringUtils.isEmpty(mqttId)){
           throw new BusinessException(ResponseCode.PARAMETER_EXCEPTION);
        }
        //那么新通过id去查询数据
        QueryWrapper<ExceptionBusiness> exceptionBusinessQueryWrapper = new QueryWrapper<>();
        exceptionBusinessQueryWrapper.eq("mqtt_id",mqttId);
        ExceptionBusiness exceptionBusiness = exceptionBusinessMapper.selectOne(exceptionBusinessQueryWrapper);
        if(null==exceptionBusiness){
            //说明成功了
            return 1;
        }
        //执行到这里 说明还有数据
        return exceptionBusiness.getStatus();
    }

    /**
     * 封装插入到数据库的数据
     *
     * @param val
     * @param queueMessage
     * @return
     */
    private ExceptionBusiness getExceptionBusiness(String val, QueueMessage queueMessage) {
        //取出这个队列名字  交换机名字  路由key   存储到 数据库中去
        ExceptionBusiness exceptionBusiness = new ExceptionBusiness();
        String msg = queueMessage.getMsg();
        JSONObject jsonObject = JSON.parseObject(msg);
        MessageHeader messageHeader = wrapMessageHeader(jsonObject);
        //然后取出这个头信息
        exceptionBusiness.setMqttId(messageHeader.getMqttId());
        exceptionBusiness.setMessage(val);
        exceptionBusiness.setStatus(0);
        exceptionBusiness.setQueueName(queueMessage.getQueueName());
        exceptionBusiness.setExchangeName(queueMessage.getExchangeName());
        exceptionBusiness.setRoutingKey(queueMessage.getRoutingKey());
        return exceptionBusiness;
    }


    /**
     * 封装头信息
     *
     * @param jsonObject
     * @return
     */
    private MessageHeader wrapMessageHeader(JSONObject jsonObject) {
        JSONObject headers = jsonObject.getJSONObject("headers");
        MessageHeader messageHeader = new MessageHeader();
        messageHeader.setMqttReceivedRetained(headers.getBoolean("mqtt_receivedRetained"));
        messageHeader.setMqttId(headers.getInteger("mqtt_id"));
        messageHeader.setTimestamp(headers.getLong("timestamp"));
        messageHeader.setMqttDuplicate(headers.getBoolean("mqtt_duplicate"));
        messageHeader.setMqttReceivedQos(headers.getInteger("mqtt_receivedQos"));
        messageHeader.setMqttReceivedTopic(headers.getString("mqtt_receivedTopic"));
        messageHeader.setId(headers.getString("id"));
        return messageHeader;
    }
}
